package com.example.KafkaDdemo.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;

@Component
@Slf4j
public class KafkaConsumer {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumer.class);

    /**
     * 如何实现重试功能？
     *  这个消费者从kafka中取出批量数据，假如取出来10条数据，在往业务方下发第三条数据的时候，发现请求业务方失败，
     *  这时候我们就要进行重试。
     *  我自己的想法是，对于该消费者休眠3s后，进行重新调用接口，此时这个消费者是阻塞状态，
     *  不能处理剩下的几条消息，在重试3次都失败后，我们认为业务方无法接受请求，可能宕机了，
     *  就要进行熔断处理。否则我们消费者会提出消费者组。
     *
     *  对于这部分数据，我要不要提交呢？
     *
     *  解决这个问题，要看一下kafka啥时候会被提出消费者组。
     *
     *  学习线路：
     *  1.了解消费者提交机制
     *  2.啥时候会踢出消费者组
     *
     * @param record
     * @param ack
     * @throws InterruptedException
     */

    // 消费监听
    @KafkaListener(topics = {"ttL_topic"}, groupId = "test-ConsumerGroupD", containerFactory = "simpleFactory")
    public void onMessage1(List<ConsumerRecord<?, ?>> record) throws InterruptedException {


        log.info("manualListenerContainerFactory 处理数据量：{}",record.size());

        for (ConsumerRecord<?, ?> consumerRecord : record) {
            log.info("value:{},key:{},partition:{}",consumerRecord.value(),consumerRecord.key(),consumerRecord.partition());
        }



        /*// 消费的哪个topic、partition的消息,打印出消息内容
        for (ConsumerRecord<?, ?> consumerRecord : record) {
            long offset = consumerRecord.offset();
            HttpHeaders headers = new HttpHeaders();
            HttpEntity httpEntity = new HttpEntity(consumerRecord.value(), headers);
            // 关注取消关注响应：{"code": 200,"msg": "success","data": null}
            ResponseEntity<String> response = restTemplate.exchange("https://callback.stg.starbucks.com.cn/isv/mop/activity/push", HttpMethod.POST, httpEntity, String.class);
            if (response.getStatusCodeValue() != 200) {
                // 休眠十分钟 重试


            }
            int value = response.getStatusCodeValue();
            System.out.println("value = " + value);

        }
        Thread.sleep(20000*10);
        ack.acknowledge();*/
    }

    // 消费监听
//    @KafkaListener(topics = {"vxtopic"})
    public void onMessage2(ConsumerRecord<?, ?> record) {
//         消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费者2：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

    // 消费监听
//    @KafkaListener(topics = {"vxtopic"})
    public void onMessage3(ConsumerRecord<?, ?> record) {


//         消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费者2：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }
}
